package com.xiaohong.canal;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.xiaohong.goods.feign.SkuFeign;
import com.xiaohong.goods.pojo.Sku;
import com.xiaohong.search.feign.SearchSkuFeign;
import com.xiaohong.search.pojo.SkuInfo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.List;
import java.util.Map;

/**
 * 读取binlog日志
 */
@Component
public class ReadBinLogService implements ApplicationRunner {

    @Value("${canal.host}")
    private String host;
    @Value("${canal.port}")
    private int port;
    @Value("${canal.username}")
    private String username;
    @Value("${canal.password}")
    private String password;
    @Value("${canal.instance}")
    private String instance;

    @Autowired
    private SkuFeign skuFeign;

    @Autowired
    private SearchSkuFeign searchSkuFeign;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        CanalConnector conn = getConn();
        while (true) {
            conn.connect();
            //订阅实例中所有的数据库和表
            //conn.subscribe(".*\\..*");
            //订阅xiaohong_goods.tb_sku
            conn.subscribe("xiaohong_goods.tb_sku");
            // 回滚到未进行ack的地方
            conn.rollback();
            // 获取数据 每次获取一百条改变数据
            Message message = conn.getWithoutAck(100);

            long id = message.getId();
            int size = message.getEntries().size();
            if (id != -1 && size > 0) {
                // 数据解析
                analysis(message.getEntries());
            }else {
                Thread.sleep(1000);
            }
            // 确认消息
            conn.ack(message.getId());
            // 关闭连接
            conn.disconnect();
        }
    }

    /**
     * 数据解析
     */
    private void analysis(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            // 只解析mysql事务的操作，其他的不解析
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN) {
                continue;
            }
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            // 解析binlog
            CanalEntry.RowChange rowChange = null;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("解析出现异常 data:" + entry.toString(), e);
            }
            if (rowChange != null) {
                // 获取操作类型
                CanalEntry.EventType eventType = rowChange.getEventType();
                // 获取当前操作所属的数据库
                String dbName = entry.getHeader().getSchemaName();
                // 获取当前操作所属的表
                String tableName = entry.getHeader().getTableName();
                // 事务提交时间
                long timestamp = entry.getHeader().getExecuteTime();
                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    dataDetails(rowData.getBeforeColumnsList(), rowData.getAfterColumnsList(), dbName, tableName, eventType, timestamp);
                    System.out.println("-------------------------------------------------------------");
                }
            }
        }
    }


    /**
     * 解析具体一条Binlog消息的数据
     *
     * @param dbName    当前操作所属数据库名称
     * @param tableName 当前操作所属表名称
     * @param eventType 当前操作类型（新增、修改、删除）
     */
    private void dataDetails(List<CanalEntry.Column> beforeColumns,
                                    List<CanalEntry.Column> afterColumns,
                                    String dbName,
                                    String tableName,
                                    CanalEntry.EventType eventType,
                                    long timestamp) {

        //根据操作类型对es进行修改
        if (eventType == CanalEntry.EventType.INSERT){
            //增加
            System.out.println("增加");
            String skuId = getSkuId(afterColumns);
            //根据id查询sku
            Sku sku = skuFeign.findById(skuId).getData();
            //ES添加
            searchSkuFeign.addOneDocument(getSkuInfo(sku));
        }
        if (eventType == CanalEntry.EventType.DELETE){
            //删除
            System.out.println("删除");
            String skuId = getSkuId(beforeColumns);
            searchSkuFeign.deleteDocument(skuId);
        }
        if (eventType == CanalEntry.EventType.UPDATE){
            //修改
            System.out.println("修改");
            String beforeId = getSkuId(beforeColumns);
            String afterId = getSkuId(afterColumns);
            //判断id是否被修改
            if (beforeId.equals(afterId)){
                //id未被修改
                //根据id查询sku
                Sku sku = skuFeign.findById(afterId).getData();
                //更新数据
                searchSkuFeign.updateDocument(getSkuInfo(sku));
            }else {
                //id被修改
                //先删除原来的数据
                searchSkuFeign.deleteDocument(beforeId);
                //添加新的数据
                //根据id查询sku
                Sku sku = skuFeign.findById(afterId).getData();
                //ES添加
                searchSkuFeign.addOneDocument(getSkuInfo(sku));
            }
        }

       /* System.out.println("数据库：" + dbName);
        System.out.println("表名：" + tableName);
        System.out.println("操作类型:" + eventType);
        if (CanalEntry.EventType.INSERT.equals(eventType)) {
            System.out.println("新增数据：");
            printColumn(afterColumns);
        } else if (CanalEntry.EventType.DELETE.equals(eventType)) {
            System.out.println("删除数据：");
            printColumn(beforeColumns);
        } else {
            System.out.println("更新数据：更新前数据--");
            printColumn(beforeColumns);
            System.out.println("更新数据：更新后数据--");
            printColumn(afterColumns);
        }
        System.out.println("操作时间：" + timestamp);*/
    }

    private void printColumn(List<CanalEntry.Column> columns) {
        for (CanalEntry.Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }

    /**
     * 获取连接
     */
    public CanalConnector getConn() {
        return CanalConnectors.newSingleConnector(new InetSocketAddress(host, port), instance, username, password);
    }

    /**
     * 获取id
     * @param columns
     * @return
     */
    private String getSkuId(List<CanalEntry.Column> columns) {
        String id = null;
        for (CanalEntry.Column column : columns) {
            if (column.getName().equalsIgnoreCase("id")) {
                id = column.getValue();
            }
        }
        return id;
    }

    /**
     * 将sku转换成skuinfo
     * @param sKu
     * @return
     */
    public SkuInfo getSkuInfo(Sku sKu){
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        SkuInfo skuInfo = new SkuInfo();
        if (sKu == null){
            return null;
        }
        if (sKu.getName() != null){
            skuInfo.setName(sKu.getName());
        }if (sKu.getId() != null){
            skuInfo.setId(sKu.getId());
        }if (sKu.getPrice() != null){
            skuInfo.setPrice(sKu.getPrice());
        }if (sKu.getNum() != null){
            skuInfo.setNum(sKu.getNum());
        }if (sKu.getStatus() != null){
            skuInfo.setStatus(sKu.getStatus());
        }if (sKu.getGmtCreate() != null){
            skuInfo.setGmtCreate(simpleDateFormat.format(sKu.getGmtCreate()));
        }if (sKu.getGmtModified() != null){
            skuInfo.setGmtModified(simpleDateFormat.format(sKu.getGmtModified()));
        }if (sKu.getSpuId() != null){
            skuInfo.setSpuId(sKu.getSpuId());
        }if (sKu.getCategoryName() != null){
            skuInfo.setCategoryName(sKu.getCategoryName());
        }if (sKu.getCategoryId() != null){
            skuInfo.setCategoryId(sKu.getCategoryId());
        }if (sKu.getBrandChineseName() != null){
            skuInfo.setBrandChineseName(sKu.getBrandChineseName());
        }if (sKu.getBrandEnglishName() != null){
            skuInfo.setBrandEnglishName(sKu.getBrandEnglishName());
        }if (sKu.getSpec() != null){
            skuInfo.setSpec(sKu.getSpec());
        }if (sKu.getSaleNum() != null){
            skuInfo.setSaleNum(sKu.getSaleNum());
        }if (sKu.getSaleNum() != null){
            skuInfo.setSaleNum(sKu.getSaleNum());
        }if (sKu.getCommentNum() != null){
            skuInfo.setCommentNum(sKu.getCommentNum());
        }if (sKu.getImages() != null){
            sKu.setImages(sKu.getImages());
        }if (sKu.getSpec() != null){
            skuInfo.setSpecMap(JSON.parseObject(sKu.getSpec(), Map.class));
        }

        return skuInfo;
    }

}


